package com.mjf.spark.day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-双value类型
 *    -合集
 *      union
 *    -交集
 *      intersect--->intersection
 *    -差集
 *      diff--->subtract
 *    -拉链
 *      zip
 */
object Spark12_Transformation_doubleValue {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark12_Transformation_doubleValue")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2: RDD[Int] = sc.makeRDD(List(4, 5, 6, 7))

    // 合集
//    val newRDD: RDD[Int] = rdd1.union(rdd2)  // 1, 2, 3, 4, 4, 5, 6, 7
    // 交集
//    val newRDD: RDD[Int] = rdd1.intersection(rdd2)    // 4
    // 差集
//    val newRDD: RDD[Int] = rdd1.subtract(rdd2)  // 1, 2, 3
//    val newRDD: RDD[Int] = rdd2.subtract(rdd1)  // 5, 6, 7
    // 拉链
    // 要求：1.分区数要保持一致 2.分区中元素个数要保持一致
    val newRDD: RDD[(Int, Int)] = rdd1.zip(rdd2)  // (1,4),(2,5),(3,6),(4,7)

    newRDD.collect().foreach(println)


    // 关闭连接
    sc. stop()

  }
}
